package cn.yyh.mq.runner;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

	@Bean
	RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
		return new RabbitAdmin(connectionFactory);
	}

	@Bean
	Queue queueFoo(RabbitAdmin rabbitAdmin) {
		Queue queue = new Queue("queue.foo", true);
		rabbitAdmin.declareQueue(queue);
		return queue;
	}

	@Bean
	Queue queueBar(RabbitAdmin rabbitAdmin) {
		Queue queue = new Queue("queue.bar", true);
		rabbitAdmin.declareQueue(queue);
		return queue;
	}
	@Bean
	Queue queueFoo1(RabbitAdmin rabbitAdmin) {
		Queue queue = new Queue("queue.foo1", true);
		rabbitAdmin.declareQueue(queue);
		return queue;
	}
	@Bean
	Queue queueFoo2(RabbitAdmin rabbitAdmin) {
		Queue queue = new Queue("queue.foo2", true);
		rabbitAdmin.declareQueue(queue);
		return queue;
	}

	@Bean
	TopicExchange topicExchange(RabbitAdmin rabbitAdmin) {
		TopicExchange topicExchange = new TopicExchange("topicExchange");
		rabbitAdmin.declareExchange(topicExchange);
		return topicExchange;
	}

	@Bean
	FanoutExchange fanoutExchange(RabbitAdmin rabbitAdmin) {
		FanoutExchange fanoutExchange = new FanoutExchange("fanoutExchange");
		rabbitAdmin.declareExchange(fanoutExchange);
		return fanoutExchange;
	}
	
	/**
	 * with("*.foo");
	 * with("queue.foo");
	 * with("queue.*"); 只有在virtual-host为/下有效
	 * 指定分发规则routingKey
	 */
	@Bean
	Binding bindingExchangeFoo(Queue queueFoo, TopicExchange topicExchange, RabbitAdmin rabbitAdmin) {
		Binding binding = BindingBuilder.bind(queueFoo).to(topicExchange).with("queue.foo");
		rabbitAdmin.declareBinding(binding);
		return binding;
	}

	@Bean
	Binding bindingExchangeBar(Queue queueBar, TopicExchange topicExchange, RabbitAdmin rabbitAdmin) {
		Binding binding = BindingBuilder.bind(queueBar).to(topicExchange).with("queue.bar");
		rabbitAdmin.declareBinding(binding);
		return binding;
	}
	@Bean
	Binding bindingExchangeFoo1(Queue queueFoo1, FanoutExchange fanoutExchange, RabbitAdmin rabbitAdmin) {
		Binding binding = BindingBuilder.bind(queueFoo1).to(fanoutExchange);
		rabbitAdmin.declareBinding(binding);
		return binding;
	}
	@Bean
	Binding bindingExchangeFoo2(Queue queueFoo2, FanoutExchange fanoutExchange, RabbitAdmin rabbitAdmin) {
		Binding binding = BindingBuilder.bind(queueFoo2).to(fanoutExchange);
		rabbitAdmin.declareBinding(binding);
		return binding;
	}

	/**
	 * 生产者使用Jackson2JsonMessageConverter转换器,对应的消费者也必须使用相同的转换器
	 * 否则消费者循环报Caused by: org.springframework.amqp.AmqpException: No method found for class [B
	 */
	@Bean
	RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
		rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
		rabbitTemplate.setConfirmCallback(new ConfirmCallback() {
			@Override
			public void confirm(CorrelationData correlationData, boolean ack, String cause) {
				System.out.println("ConfirmCallback:");
				System.out.println(ack);
				System.out.println(cause);
				System.out.println(correlationData);
			}
		});
		rabbitTemplate.setReturnCallback(new ReturnCallback() {
			@Override
			public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
				System.out.println("ReturnCallback:");
				System.out.println(message.getBody().toString());
			}
		});
		return rabbitTemplate;
	}

	/**
	 * 消费者使用Jackson2JsonMessageConverter转换器
	 */
	@Bean
	public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);
		factory.setMessageConverter(new Jackson2JsonMessageConverter());
		return factory;
	}
}